package jm.easyconfig;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.UUID;

import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.util.ClassUtils;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;

public class RabbitmqUpdater extends ConfigUpdater implements Consumer {
	static long startupTime = CfgHelper.unixMsOfNow();
	CfgContextEnvironmentAware ctxAware;
	MySqlConfigSource configSource;
	Channel channel;

	String exchangeName() {
		return "easyconfig:" + CfgContextEnvironmentAware.clientUniqueName(CfgContextEnvironmentAware.environment);
	}

	String queueName() {
		return queueName(CfgContextEnvironmentAware.nodeName());
	}

	String queueName(String nodeName) {
		return "easyconfig:" + CfgContextEnvironmentAware.environment.getProperty("easyconfig.application-name", "")
				+ CfgContextEnvironmentAware.environment.getProperty("easyconfig.client", "main") + ":" + nodeName;
	}

	Channel createChannel() throws Throwable {
		ConfigurableEnvironment env = CfgContextEnvironmentAware.environment;

		ConnectionFactory factory = null;
		Channel channel = null;

		if (env.containsProperty("easyconfig.rabbitmq.addresses")) {
			factory = new ConnectionFactory();
//			factory.setUri("amqp://guest:guest@localhost:5672/%2F");
			factory.setHost(env.getProperty("easyconfig.rabbitmq.addresses"));
			factory.setPort(env.getProperty("easyconfig.rabbitmq.port", Integer.class, 5672));
			factory.setUsername(env.getProperty("easyconfig.rabbitmq.username", "guest"));
			factory.setPassword(env.getProperty("easyconfig.rabbitmq.password", "guest"));
			factory.setVirtualHost(env.getProperty("easyconfig.rabbitmq.virtual-host", "/"));
			channel = factory.newConnection().createChannel();

			CfgHelper.logger.info("Create rabbitmq channel with {}", env.getProperty("easyconfig.rabbitmq.addresses"));
		}

		if (channel == null
				&& ClassUtils.isPresent("org.springframework.amqp.rabbit.connection.AbstractConnectionFactory",
						this.getClass().getClassLoader())) {

			String[] names = CfgContextEnvironmentAware.applicationContext
					.getBeanNamesForType(AbstractConnectionFactory.class, false, false);

			if (names.length > 1) {
				CfgHelper.logger.warn(
						"Find more than one[{}] AbstractConnectionFactory in spring applicationContext. config specific one with properties of easyconfig.rabbitmq.*.",
						names.length);
			}

			for (int i = 0; i < names.length; i++) {
				try {
					AbstractConnectionFactory sf = (AbstractConnectionFactory) CfgContextEnvironmentAware.applicationContext
							.getBean(names[i]);

					channel = sf.getRabbitConnectionFactory().newConnection().createChannel();
//						channel = sf.createConnection().createChannel(false);

					CfgHelper.logger.info(
							"Succeeded to create rabbitmq channel with AbstractConnectionFactory[{}] {} in spring application context.",
							i, names[i]);
					break;

				} catch (Throwable ex) {
					CfgHelper.logger.error(
							"Failed to create rabbitmq channel with AbstractConnectionFactory[{}] {} in spring application context().",
							i, names[i]);
				}
			}

		}

		if (channel == null && env.containsProperty("spring.rabbitmq.addresses")) {
			factory = new ConnectionFactory();
			factory.setHost(env.getProperty("spring.rabbitmq.addresses"));
			factory.setPort(env.getProperty("spring.rabbitmq.port", Integer.class, 5672));
			factory.setUsername(env.getProperty("spring.rabbitmq.username", "guest"));
			factory.setPassword(env.getProperty("spring.rabbitmq.password", "guest"));
			factory.setVirtualHost(env.getProperty("spring.rabbitmq.virtual-host", "/"));

			channel = factory.newConnection().createChannel();

			CfgHelper.logger.info("Create rabbitmq channel with {}", env.getProperty("spring.rabbitmq.addresses"));
		}

		if (channel == null) {
			CfgHelper.logger.error(
					"rabbitmq configuration is not found. please config it with properties of easyconfig.rabbitmq.*");
		}

		return channel;
	}

	@Override
	public void init(MySqlConfigSource configSource, CfgContextEnvironmentAware ctxAware) {
		this.configSource = configSource;
		this.ctxAware = ctxAware;

		List<ConfigItemCore> list = configSource.queryAfterTime(new Holder<Long>(0L));

		ctxAware.profilesHelper.replaceEnvProperties(list);

		try {
			channel = createChannel();

			channel.exchangeDeclare(exchangeName(), BuiltinExchangeType.FANOUT);

			final String queue = channel.queueDeclare(queueName(), false, true, true, null).getQueue();// channel.queueDeclare().getQueue();

			channel.queueBind(queue, exchangeName(), "");

			channel.basicConsume(queue, true, this);

		} catch (Throwable ex) {
			this.close();

			CfgHelper.logger.info("error when to init amqp refresher.{}", ex.getMessage());
			ex.printStackTrace();
		}
	}

	@Override
	public void handleConsumeOk(String consumerTag) {
		// TODO Auto-generated method stub

	}

	@Override
	public void handleCancelOk(String consumerTag) {
		// TODO Auto-generated method stub

	}

	@Override
	public void handleCancel(String consumerTag) throws IOException {
		// TODO Auto-generated method stub

	}

	Object callMonitor = new Object();
	Object returnMonitor = new Object();
	String correlationId;
	String correlationReturn;

	public Object inspect(String targeNode, String itemCode) {
		String corrid = UUID.randomUUID().toString();
		BasicProperties props = new BasicProperties.Builder().type("inspect").correlationId(corrid)
				.replyTo(this.queueName()).build();

		synchronized (callMonitor) {
			try {
				channel.basicPublish("", this.queueName(targeNode), props, itemCode.getBytes(StandardCharsets.UTF_8));
			} catch (IOException e1) {
				throw new ApplicationException(e1);
			}

			int count = 500;
			while (count > 0) {
				count--;
				synchronized (returnMonitor) {
					if (!corrid.equals(correlationId)) {
						try {
							Thread.sleep(10);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						continue;
					}
					return CfgHelper.jsonRead(correlationReturn, PropDetail.class);
				}
			}

			throw new ApplicationException(
					String.format("timeout to call node of %s with command of %s", targeNode, itemCode));
		}
	}

	@Override
	public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
			byte[] messagebody) throws IOException {
		try {
			CfgHelper.logger.debug("consumerTag:{},envelope:{},properties:{}", consumerTag, envelope, properties);

			if ("inspect".equals(properties.getType())) {

				String itemCode = new String(messagebody, StandardCharsets.UTF_8);

				Object ret = CfgContextEnvironmentAware.getSharedInstance().inspect(itemCode);

				AMQP.BasicProperties replyProps = new BasicProperties.Builder()
						.correlationId(properties.getCorrelationId()).type("return").build();

				channel.basicPublish("", properties.getReplyTo(), replyProps,
						CfgHelper.jsonWrite(ret).getBytes(StandardCharsets.UTF_8));
				return;
			} else if ("return".equals(properties.getType())) {
				synchronized (returnMonitor) {
					correlationReturn = new String(messagebody, StandardCharsets.UTF_8);
					correlationId = properties.getCorrelationId();
				}
				return;
			}

			String jsonText = new String(messagebody, StandardCharsets.UTF_8);
			MessageBody body = CfgHelper.jsonRead(jsonText, MessageBody.class);
			switch (body.command) {
			case publishEveryone:
				List<ConfigItemCore> list = Arrays.asList(body.payloadAs(ConfigItemCore[].class));
				CfgContextEnvironmentAware.applicationContext.publishEvent(new ConfigRefreshEvent(this, list, false));
				break;
			case publishSome:
				if (body.to.contains(CfgContextEnvironmentAware.nodeName())) {
					List<ConfigItemCore> list2 = Arrays.asList(body.payloadAs(ConfigItemCore[].class));
					CfgContextEnvironmentAware.applicationContext
							.publishEvent(new ConfigRefreshEvent(this, list2, false));
				}
				break;
			case nodesGather:
				ackNodesGather(body);
				break;
			case ackNodesGather:
				dealAckNodesGather(body);
				break;
			default:
				CfgHelper.logger.error("invalid message command.{}", body.command);
			}

			CfgHelper.logger.info("[{}] Received {} from [{}].", CfgContextEnvironmentAware.nodeName(), jsonText,
					body.from);

		} catch (Throwable ex) {
			CfgHelper.logger.error("error when to receive amqp message.{}", ex.getMessage());
			ex.printStackTrace();
		}
	}

	@Override
	public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
		// TODO Auto-generated method stub

	}

	@Override
	public void handleRecoverOk(String consumerTag) {
		// TODO Auto-generated method stub

	}

	@Override
	public void close() {
		try {
			if (this.channel != null) {
				this.channel.close();

				if (this.channel.getConnection() != null) {
					this.channel.getConnection().close();
				}
			}
		} catch (Throwable ex) {
			CfgHelper.logger.info("error when to close amqp.{}", ex.getMessage());
			ex.printStackTrace();
		}
	}

	@Override
	public void publish(List<ConfigItemCore> updatedResult, List<String> nodeList) {
		try {
			MessageBody body = new MessageBody();

			if (nodeList != null) {
				body.withCommand(MessageCommand.publishSome).withTo(nodeList)
						.withFrom(CfgContextEnvironmentAware.nodeName());
			} else
				body.withCommand(MessageCommand.publishEveryone).withFrom(CfgContextEnvironmentAware.nodeName());

			body.withPayload(updatedResult);

			channel.basicPublish(exchangeName(), "", null, CfgHelper.jsonWrite(body).getBytes(StandardCharsets.UTF_8));

		} catch (IOException ex) {
			CfgHelper.logger.error("error when to publish amqp message.{}", ex.getMessage());
			ex.printStackTrace();
		}
	}

	public void nodesGather() {
		try {
			// step1. initialize node data bag to gather
			ConfigItem item = this.configSource.loadConfigItem(SysProps.APPLICATION_NODES, SysProps.applicationNodes);
			ApplicationNodes nodes = new ApplicationNodes().gatheredBy(CfgContextEnvironmentAware.nodeName());
			item.withValue(CfgHelper.jsonWrite(nodes));
			this.configSource.updateConfigItem(item);

			// step 2. send message
			MessageBody body = new MessageBody().withCommand(MessageCommand.nodesGather)
					.withFrom(CfgContextEnvironmentAware.nodeName()).withPayload(System.nanoTime());
			channel.basicPublish(exchangeName(), "", null, CfgHelper.jsonWrite(body).getBytes(StandardCharsets.UTF_8));

		} catch (Throwable ex) {
			ex.printStackTrace();
			throw new RuntimeException(ex);
		}
	}

	public void ackNodesGather(MessageBody receivedBody) throws Throwable {

		MessageBody bodyToSend = new MessageBody().withCommand(MessageCommand.ackNodesGather)
				.withFrom(CfgContextEnvironmentAware.nodeName())
				.withPayload(receivedBody.payloadAs(Long.class), startupTime, ctxAware.profilesHelper.profiles);

		channel.basicPublish("", queueName(receivedBody.from), null,
				CfgHelper.jsonWrite(bodyToSend).getBytes(StandardCharsets.UTF_8));
	}

	public void dealAckNodesGather(MessageBody receivedBody) {

		ConfigItem item = this.configSource.loadConfigItem(SysProps.APPLICATION_NODES, SysProps.applicationNodes);
		ApplicationNodes nodes = CfgHelper.jsonRead(item.item_value, ApplicationNodes.class);

		Object[] payLoad = receivedBody.payloadAs(Object[].class);
		String[] profiles = CfgHelper.jsonRead(CfgHelper.jsonWrite(payLoad[2]), String[].class);
		NodeRecord that = new NodeRecord().withName(receivedBody.from)
				.withAckUsed((System.nanoTime() - Long.parseLong(payLoad[0].toString())) / 1000000)
				.withUptime(Long.parseLong(payLoad[1].toString())).withProfiles(profiles);

		nodes.addNode(that);

		item.item_value = CfgHelper.jsonWrite(nodes);

		this.configSource.updateConfigItem(item);
	}

	static class ApplicationNodes {
		public String lastGatheredBy;
		public long lastGatheredTime;
		public NodeRecord[] list;

		public ApplicationNodes() {
			super();
			this.lastGatheredTime = CfgHelper.unixMsOfNow();
			this.list = new NodeRecord[0];
		}

		ApplicationNodes gatheredBy(String gatheredBy) {
			this.lastGatheredBy = gatheredBy;
			return this;
		}

		boolean addNode(NodeRecord node) {

			ArrayList<NodeRecord> lst = new ArrayList<NodeRecord>(this.list.length + 1);

			for (int i = 0; i < this.list.length; i++) {
				int c = list[i].name.compareTo(node.name);
				if (c == 0)
					return false;
				else if (c > 0) {
					lst.add(node);
				}

				lst.add(list[i]);
			}

			if (this.list.length == 0 || this.list[this.list.length - 1].name.compareTo(node.name) < 0)
				lst.add(node);

			this.list = lst.toArray(new NodeRecord[lst.size()]);

			return true;
		}
	}

	static class NodeRecord {
		public String name;
		public long uptime;// time to startup this application instance.
		// if lastGatherTime > ackTime then node is not available;
		public long ackTime;
		public long ackUsed;// acknowledge time duration elapsed
		public String[] profiles;

		public NodeRecord() {
			this.ackTime = CfgHelper.unixMsOfNow();
		}

		NodeRecord withName(String name) {
			this.name = name;
			return this;
		}

		NodeRecord withUptime(long uptime) {
			this.uptime = uptime;
			return this;
		}

		NodeRecord withAckUsed(Long ackUsed) {
			this.ackUsed = ackUsed;
			return this;
		}

		NodeRecord withProfiles(String[] profiles) {
			this.profiles = profiles;
			return this;
		}
	}

	static enum MessageCommand {
		publishEveryone, publishSome, nodesGather, ackNodesGather
	}

	static class MessageBody {
		public MessageCommand command;
		public String from;
		public List<String> to;
		public String payload;

		MessageBody withCommand(MessageCommand command) {
			this.command = command;
			return this;
		}

		MessageBody withFrom(String from) {
			this.from = from;
			return this;
		}

		MessageBody withTo(List<String> to) {
			this.to = to;
			return this;
		}

		MessageBody withPayload(Object payload) {
			this.payload = CfgHelper.jsonWrite(payload);
			return this;
		}

		MessageBody withPayload(Object... payload) {
			this.payload = CfgHelper.jsonWrite(payload);
			return this;
		}

		<T> T payloadAs(Class<T> clazz) {
			return CfgHelper.jsonRead(this.payload, clazz);

		}
	}
}
